﻿
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using InCom.Logic.Process.Translation;

namespace InCom.Core.TerminalSend
{
    using inCom.Logs;
    using InCom.Core.TerminalSend.MappedFile;
    using InCom.Entities.ProtocolEntity;
    using InCom.SyncStatus;
    using InCom.Utilities;
    using InCom.Utilities.Pool;
    using InCom.Utilities.Protocol;
    using System.Collections.Concurrent;
    using ApiRealtimeUpEntity = InCom.API.Protocols.Entitys.RealtimeUpEntity;
    /// <summary>
    /// 用于发送给incom.api的实时数据转发缓冲区
    /// 实际测试,每秒处理10000条数据,超过缓冲区将增加.
    /// </summary>
    public class RealDataTransmissionSendBuffer : IPool
    {



        #region 构造函数与单例

        /// <summary>
        /// 用于每一个节点一个的发送线程
        /// </summary>
        internal ConcurrentDictionary<string, RealSyncPointData> dicTagTasks = new ConcurrentDictionary<string, RealSyncPointData>();
        /// <summary>
        /// 构造函数
        /// </summary>
        protected RealDataTransmissionSendBuffer()
        {
            //初始化池对像
            InitalizePool();

            //找到生成协议解析方法 
            //查找到已经存在的协议处理逻辑单元方法实例
            translation = new TerminalRealtimeUpEntityTranslation();
            parse = InCom.API.Protocols.PluginModulesManager.Instance.GetTcpParse((ushort)API.Protocols.APIProtocolType.REALTIMEDATA);


            //初始化实时数据解析多线程
            Enumerable.Range(0, 2).ToList().ForEach((index) =>
                                                        {
                                                            //线程数据对像
                                                            RealSyncPointData realsyncData = new RealSyncPointData();
                                                            //创建线程对像
                                                            TaskEx task = new TaskEx(SendToSyncPoint, realsyncData, new System.Threading.CancellationTokenSource()) { AutoReRun = true, StartTime = DateTime.Now, ThreadName = "RealDataTransmissionSendBuffer ,从incom将实时数据.发送到其它数据同步节点的线程" + index };
                                                            //设置线程处理对像
                                                            realsyncData.Task = task;
                                                            ParseTasks.TryAdd(index, realsyncData);
                                                            task.Start();
                                                            ThreadManager.Instance.Add(task);
                                                        });


            //初始化终端注册发送实时数据功能
            InCom.Core.SyncInteractive.SyncDataInteractiveSupervise.Instance.OnTerminalRegistAction = realDesk.RegistTerminal;

        }

        private static readonly RealDataTransmissionSendBuffer instance = new RealDataTransmissionSendBuffer();
        /// <summary>
        /// 数据同步与incom交互主要处理功能,静态访问单例
        /// </summary>
        public static RealDataTransmissionSendBuffer Instance
        {
            get { return instance; }
        }

        #endregion


        /// <summary>
        /// 重新开始记数
        /// </summary>
        internal DateTime startTime = DateTime.Now;

        internal long sendCount = 0;
        /// <summary>
        /// 发送数据字节统计
        /// </summary>
        public long SendCount
        {
            get { return sendCount; }
            set { sendCount = value; }
        }


        /// <summary>
        /// 创建一个用于发送的数据对像
        /// </summary>
        /// <param name="syncPoint">数据节点</param>
        /// <returns></returns>
        protected RealSyncPointData TagBufferCreate(SyncPoint syncPoint)
        {
            RealSyncPointData buf = null;
            lock (dicTagTasks)
            {
                buf = new RealSyncPointData()
                {
                    TagName = syncPoint.TagName,
                    SyncPoint = syncPoint
                };
                if (dicTagTasks.TryAdd(syncPoint.TagName, buf))
                {         
                    buf.Task = new TaskEx(SendToDisk, buf, new System.Threading.CancellationTokenSource());
                    //开始发送线程
                    buf.Task.Start();
                }
            }

            return buf;
        }

        /// <summary>
        /// 实时数据池
        /// </summary>
        protected BasePool<RealDataToSyncPointEntity> realPool = new BasePool<RealDataToSyncPointEntity>(1000, 5000, false);

        /// <summary>
        /// 将需要发送的实时数据对像添加到发送队列缓冲区
        /// </summary>
        /// <param name="readdata">需要发送的实时数据对像</param>
        /// <param name="syncPoint">需要发送到的数据同步节点</param>
        /// <param name="transmission">需要发送的目标节点</param>
        public void Push(RealtimeUpEntity realdata, Dictionary<string, SyncPoint> transmission)
        {
            var item = realPool.Pop();
            if (item == null)
            {
                item = new RealDataToSyncPointEntity() { Pool = realPool };
            }
            item.Protocol = realdata;
            // item.SyncPoint = syncPoint;
            item.TerminalCode = realdata.TerminalCode;
            item.Transmission = transmission;
#if old
             //用于等待解析的数据
            lock (sendQueue)
            {
                //加入到发送队列缓冲区
                sendQueue.Enqueue(item);
            }
#endif
            //取模,每一个将需要解析的数据转到多个解析线程中
            //跟据终端号分配线程
            int index = (int)(item.TerminalCode % ParseTasks.Count);
            //每一个节点一个线程
            RealSyncPointData waitParseTask = ParseTasks[index];
            //添加数据
            lock (waitParseTask.Queuedata)
            {
                waitParseTask.Queuedata.Enqueue(item);
            }

        }



        /// <summary>
        /// 将需要发送的实时数据对像添加到发送队列缓冲区
        /// </summary>
        /// <param name="readdata">需要发送的实时数据对像</param>
        /// <param name="syncPoint">需要发送到的数据同步节点</param>
        /// <param name="sendBytes">需要发送的数据</param>
        protected void Push(ApiRealtimeUpEntity realdata, SyncPoint syncPoint, byte[] sendBytes)
        {

            var item = Pop();
            if (item == null)
            {
                item = new RealDataToSyncPointEntity() { Pool = this };
            }
            // item.Protocol = realdata;
            item.SyncPoint = syncPoint;
            //设置发送值
            item.ToSendBytes = sendBytes;
            item.TerminalCode = realdata.TerminalCode;
            //实时数据发送对像
            RealSyncPointData realsyncdata = null;
            //每一个节点一个线程
            lock (dicTagTasks)
            {
                dicTagTasks.TryGetValue(syncPoint.TagName, out realsyncdata);
                if (realsyncdata == null)
                {
                    realsyncdata = TagBufferCreate(syncPoint);
                }
            }
            //添加数据
            lock (realsyncdata.Queuedata)
            {
                realsyncdata.Queuedata.Enqueue(item);
            }


        }

        /// <summary>
        /// 将需要发送的实时数据对像添加到发送队列缓冲区
        /// </summary>
        /// <param name="readdata">需要发送的实时数据对像</param>
        /// <param name="syncPoint">需要发送到的数据同步节点</param>
        protected void Push(RealtimeUpEntity realdata, SyncPoint syncPoint)
        {

            var item = Pop();
            if (item == null)
            {
                item = new RealDataToSyncPointEntity() { Pool = this };
            }
            item.Protocol = realdata;
            item.SyncPoint = syncPoint;
            item.TerminalCode = realdata.TerminalCode;
            //实时数据发送对像
            RealSyncPointData realsyncdata = null;
            //每一个节点一个线程
            lock (dicTagTasks)
            {
                dicTagTasks.TryGetValue(syncPoint.TagName, out realsyncdata);
                if (realsyncdata == null)
                {
                    realsyncdata = TagBufferCreate(syncPoint);
                }
            }
            //添加数据
            lock (realsyncdata.Queuedata)
            {
                realsyncdata.Queuedata.Enqueue(item);
            }

            ////需要持久化发送数据
            //if (syncPoint.SyncPointType == SyncPointType.RealDataCalculate)
            //{
            //    //加入持久化发送数据队列
            //    lock (diskQueue)
            //    {
            //        diskQueue.Enqueue(item);
            //    }
            //}
            //else
            //{
            //    //实时发送最新数据
            //    lock (sendQueue)
            //    {
            //        //加入到发送队列缓冲区
            //        sendQueue.Enqueue(item);
            //    }
            //}

            //表示可以发送数据
            // hand.Set();
        }

        //异步handler
        private System.Threading.AutoResetEvent hand = new System.Threading.AutoResetEvent(false);


        /// <summary>
        /// 发送时实数据到磁盘中,然后再从磁盘中读取数据发送到不同节点
        /// </summary>
        internal RealSendBytesToDisk realDesk = new RealSendBytesToDisk();

        #region 发送对像池
        /// <summary>
        /// 池队列
        /// </summary>
        protected Queue<RealDataToSyncPointEntity> poolQueue = new Queue<RealDataToSyncPointEntity>(3000);

        /// <summary>
        /// 获取一个对像
        /// </summary>
        /// <returns></returns>
        protected RealDataToSyncPointEntity Pop()
        {
            lock (poolQueue)
            {
                return poolQueue.Count > 0 ? poolQueue.Dequeue() : new RealDataToSyncPointEntity() { Pool = this };
            }
        }

        /// <summary>
        /// 还池
        /// </summary>
        /// <param name="bakcObject"></param>
        public void Push(object bakcObject)
        {
            RealDataToSyncPointEntity obj = (RealDataToSyncPointEntity)bakcObject;
            obj.Protocol = null;
            obj.SyncPoint = null;
            obj.TerminalCode = 0;
            obj.ToSendBytes = null;
            obj.Transmission = null;
            //20000个以内则保存在池里,超过20000个则不做处理.等待回收
            lock (poolQueue)
            {
                if (poolQueue.Count < 20000)
                {
                    //还池
                    poolQueue.Enqueue(obj);
                }
            }


        }



        /// <summary>
        /// 初始化池
        /// </summary>
        protected void InitalizePool()
        {
            //初始化池对像
            System.Linq.Enumerable.Range(1, 200).ToList().ForEach((o) =>
                                                                      {
                                                                          poolQueue.Enqueue(new RealDataToSyncPointEntity() { Pool = this });
                                                                      });
        }

        #endregion


        /// <summary>
        /// 用于统计的记数器,
        /// </summary>
        /// <param name="obj"></param>
        protected void Statistical(RealDataToSyncPointEntity obj)
        {
            //代码没有编写

        }

        /// <summary>
        /// 实时数据转换用接口
        /// </summary>
        protected TerminalRealtimeUpEntityTranslation translation = null;

        /// <summary>
        /// 解析方法
        /// </summary>
        protected ITcpParse parse = null;

        #region 解析协议
        /// <summary>
        /// 发送绥冲区的队列
        /// </summary>
        internal Queue<RealDataToSyncPointEntity> sendQueue = new Queue<RealDataToSyncPointEntity>(3000);





        #region 多线程解析

        /// <summary>
        /// 单线程解析数据
        /// </summary>
        private void SendToSyncPoint(object o)
        {
            //强制类型转换
            RealSyncPointData realparse = (RealSyncPointData)o;

            if (translation == null)
            {

                LogProvider.Create().Write(EventType.Error, string.Format("incom->incom.api 实时数据互转类型没有加载,没有找到对应的转换方法 RealDataTransmissionSendBuffer.SendToSyncPoint()"));
                return;
            }
            if (parse == null)
            {
                LogProvider.Create().Write(EventType.Error, string.Format("incom.api 实时数据互解析方法没有加载,RealDataTransmissionSendBuffer.SendToSyncPoint()"));
                return;
            }

            while (true)
            {
                try
                {
                    //循环处理数据
                    CycleWorker(realparse.Queuedata);
                    //hand.WaitOne();

                }
                catch (Exception ex)
                {
                    LogProvider.Create().Write(EventType.Warning, string.Format("方法RealDataTransmissionSendBuffer.CycleWorker(),{0}", ex));
                }
            }
        }

        /// <summary>
        /// 循环处理数据
        /// </summary>
        private void CycleWorker(Queue<RealDataToSyncPointEntity> parseQueue)
        {

            RealDataToSyncPointEntity obj = null;
            lock (parseQueue)
            {
                if (parseQueue.Count > 0)
                {
                    obj = parseQueue.Dequeue();
                }
            }
            if (obj == null)
            {
                System.Threading.Thread.Sleep(5);
                return;
            }
            //如果没有找到需要转发的目标则,转发数据无效
            if (obj.Transmission != null && obj.Transmission.Count <1)
            {
                return;
            }

            //从池中取对像这样提升效率
            ApiRealtimeUpEntity real = ruRealpool.Pop();
            var apireal = (real == null ? new ApiRealtimeUpEntity() : real);

            //先转换为incom.api数据
            var apiProtocol = translation.GetInComAPIEntity(obj.Protocol, apireal);
            //再进行数据转换
            if ((apiProtocol as API.Protocols.Entitys.RealtimeUpEntity) != null)
            {

                    //这行代码是瓶颈,把二进制数据构造成buffer
                    byte[] sendBuff = parse.Create(apiProtocol);

                    //开始添加进入处理队列
                    foreach (var item in obj.Transmission.Values.ToArray())
                    {
                        Push(apireal, item, sendBuff);
                    }

            }
            else
            {
                LogProvider.Create().Write(EventType.Warning, string.Format("incom 实时数据转发到incom.api 时出现问题, 未能将 incom.api 类型转换为 RealtimeUpEntity,  方法RealDataTransmissionSendBuffer.CycleWorker()"));
            }

            //还池
            ruRealpool.Push(apireal);
            obj.Transmission = null;
            obj.SyncPoint = null;
            obj.Protocol = null;
            //将使用过后的队列还池
            realPool.Push(obj);

        }

        //操作步骤
        //1.对终端号取模,拆分成3个线程进行解析
        /// <summary>
        /// 解析线程
        /// </summary>
        internal System.Collections.Concurrent.ConcurrentDictionary<int, RealSyncPointData> ParseTasks = new ConcurrentDictionary<int, RealSyncPointData>();



        #endregion


        #endregion


        /// <summary>
        /// 用于解析的实时数据池,提升效率
        /// </summary>
        BasePool<ApiRealtimeUpEntity> ruRealpool = new BasePool<ApiRealtimeUpEntity>(20, 100, false);

        #region  发送即时数据; 持久化数据->写磁盘->发送

        /// <summary>
        /// 持久化发送数据
        /// </summary>
        private void SendToDisk(object o)
        {
            //强制类型转换
            RealSyncPointData realData = (RealSyncPointData)o;

            if (translation == null)
            {

                LogProvider.Create().Write(EventType.Error,
                                           string.Format(
                                               "incom->incom.api 实时数据互转类型没有加载,没有找到对应的转换方法 RealDataTransmissionSendBuffer.SendToSyncPoint()"));
                return;
            }
            if (parse == null)
            {
                LogProvider.Create().Write(EventType.Error,
                                           string.Format(
                                               "incom.api 实时数据互解析方法没有加载,RealDataTransmissionSendBuffer.SendToSyncPoint()"));
                return;
            }

            while (true)
            {
                try
                {
                    //循环处理数据
                    CycleToDisk(realData.Queuedata);
                    //hand.WaitOne();

                }
                catch (Exception ex)
                {
                    LogProvider.Create().Write(EventType.Warning,
                                               string.Format("方法RealDataTransmissionSendBuffer.CycleWorker(),{0}", ex));
                }
            }
        }


        private DateTime oldtime = DateTime.MinValue;
        /// <summary>
        /// 循环写入磁盘
        /// </summary>
        private void CycleToDisk(Queue<RealDataToSyncPointEntity> queue)
        {



            //将要发送的数据
            List<RealDataToSyncPointEntity> willSends = new List<RealDataToSyncPointEntity>();
            //持久化数据
            List<RealDataToSyncPointEntity> sendTodisk = new List<RealDataToSyncPointEntity>();
            //即时数据
            List<RealDataToSyncPointEntity> sendToReal = new List<RealDataToSyncPointEntity>();
            #region 读取队列数据
            //当前读取字节长度
            int curReadCount = 0;
            //读取到磁盘
            int curToDisk = 0;
            //读取发送到非持久化节点
            int curToReal = 0;
            //单次最大读取长度
            int maxReadlen = 1024 * 120;
            for (; curReadCount < maxReadlen && queue.Count > 0; )
            {
                RealDataToSyncPointEntity obj = null;
                lock (queue)
                {
                    if (queue.Count > 0)
                    {
                        obj = queue.Dequeue();
                    }
                }
                willSends.Add(obj);
                if (obj.ToSendBytes != null)
                {
                    curReadCount += obj.ToSendBytes.Length;
                    //到磁盘持久化节点
                    if (obj.SyncPoint.SyncPointType == SyncPointType.RealDataCalculate)
                    {
                        sendTodisk.Add(obj);
                        //添加到集合
                        curToDisk += obj.ToSendBytes.Length;
                    }
                    else
                    {
                        sendToReal.Add(obj);
                        //添加到读取非持久化节点
                        curToReal += obj.ToSendBytes.Length;
                    }
                }
            }

            #endregion
            if (willSends.Count == 0)
            {
                System.Threading.Thread.Sleep(5);
                return;
            }

            #region 持久化发送数据
            //为了在持久化发送线程中计数正确采用一个循环
            foreach (var diskItem in sendTodisk)
            {
                if (diskItem.ToSendBytes != null)
                {
                    //将实时数据发送到磁盘中
                    realDesk.Write(diskItem.ToSendBytes, diskItem.SyncPoint);
                }
            }

            //持久化数据,然后一次性写入到磁盘
            //byte[] diskArr = GetSendBytes(curToDisk, sendTodisk);

            #endregion

            #region 将非持久化数据按节点分组
            foreach (var realobjs in sendToReal.ToLookup(p => p.SyncPoint.TagName))
            {
                var list = realobjs.ToList();
                int len = list.Sum(p => p.ToSendBytes.Length);
                //非持久化即时数据
                byte[] realArr = GetSendBytes(len, list);
                //发送即时数据
                if (realArr.Length > 0)
                {
                    //将数据发送到指令的 数据同步节点,不在日志文件中记录
                    SyncPointManager.Instance.Send(list[0].SyncPoint, realArr, false);
                    //增加数值
                    System.Threading.Interlocked.Add(ref sendCount, list.Count);
                }
            }
            #endregion

            //还池
            foreach (var item in willSends)
            {
                //进行发送统计
                Statistical(item);
                //将使用过后的队列还池
                Push(item);
            }

            willSends.Clear();
            sendTodisk.Clear();
            sendToReal.Clear();


        }
        /// <summary>
        /// 获取将要发送的数据
        /// </summary>
        /// <param name="len">数据长度</param>
        /// <param name="willSends">总数据记录</param>
        /// <returns></returns>
        protected byte[] GetSendBytes(int len, List<RealDataToSyncPointEntity> willSends)
        {
            byte[] willSendbytes = new byte[len];
            //开始复制数据
            int perIndex = 0;
            foreach (var arr in willSends)
            {
                if (arr.ToSendBytes != null)
                {
                    Buffer.BlockCopy(arr.ToSendBytes, 0, willSendbytes, perIndex, arr.ToSendBytes.Length);
                    perIndex += arr.ToSendBytes.Length;
                }
            }
            return willSendbytes;
        }

        #endregion
    }

    /// <summary>
    /// 实时数据节点,对应的线程对像
    /// </summary>
    public class RealSyncPointData
    {

        /// <summary>
        /// 缓存用队列
        /// </summary>
        internal Queue<RealDataToSyncPointEntity> Queuedata = new Queue<RealDataToSyncPointEntity>(3000);

        /// <summary>
        /// 数据数据对像
        /// </summary>
        public object UserToKen { get; set; }

        /// <summary>
        /// 标识
        /// </summary>
        public string TagName { get; set; }

        /// <summary>
        /// 对应的任务对像
        /// </summary>
        public TaskEx Task { get; set; }


        /// <summary>
        /// 数据同步节点类型
        /// </summary>
        public SyncPoint SyncPoint { get; set; }

    }
}
